package com.ksyun.campus.metaserver.services;

import cn.hutool.core.date.DateUtil;
import com.ksyun.campus.common.constant.ZkConstant;
import com.ksyun.campus.common.constant.api.DataServerApi;
import com.ksyun.campus.common.domain.*;
import com.ksyun.campus.common.util.ByteConvertUtil;
import com.ksyun.campus.metaserver.util.FSHttpUtils;
import org.apache.curator.framework.CuratorFramework;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.ResponseEntity;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.web.bind.annotation.*;

import java.util.*;
import java.util.stream.Collectors;

@Service
public class MetaService {


    @Autowired
    private CuratorFramework curatorFramework;


    @Autowired
    private RegistService registService;


    /**
     * 获取数据集群
     * @return
     */
    public List<DataServer> getDataList(){
        Set<Map.Entry<String, DataServer>> entries = DataDiscovery.dataServerStore.entrySet();
        List<DataServer> dataServers = entries.stream().map(Map.Entry::getValue)
                .collect(Collectors.toList());
        return dataServers;
    }


    public DataServer pickDataServer() {
        // todo 通过zk内注册的ds列表，选择出来一个ds，用来后续的wirte
        Set<Map.Entry<String, DataServer>> entries = DataDiscovery.dataServerStore.entrySet();
        List<DataServer> dataServers = entries.stream().map(Map.Entry::getValue)
                .collect(Collectors.toList());
        if (dataServers.size() == 0) {
            return null;
        }
        //排序,选择使用容量最小的
        DataServer dataServer = dataServers.stream().min(Comparator.comparingInt(DataServer::getUseCapacity)).get();
        return dataServer;
    }

    /**
     * 查看文件元信息
     */
    public ResponseEntity stats(String fileSystem, String path) {
        String metapath = String.format("%s/%s%s", ZkConstant.META_DATA_PATH, fileSystem, path);
        try {
            StatInfo statInfo = getStatInfo(metapath);
            return ResponseEntity.ok(statInfo);
        } catch (Exception e) {
            return ResponseEntity.notFound().build();
        }
    }

    /**
     * 创建文件
     */
    public ResponseEntity createFile(String fileSystem, String path) {
        try {
            String[] split = path.split("/");
            mkdirParent(split, String.format("%s/%s", ZkConstant.META_DATA_PATH, fileSystem), "", 0);
            String metapath = String.format("%s/%s%s", ZkConstant.META_DATA_PATH, fileSystem, path);
            StatInfo statInfo = new StatInfo();
            statInfo.setPath(path);
            statInfo.setMtime(DateUtil.date().getTime());
            statInfo.setType(FileType.File);
            if (curatorFramework.checkExists().forPath(metapath) == null) {
                curatorFramework.create().forPath(metapath, ByteConvertUtil.toByte(statInfo));
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return ResponseEntity.ok().build();
    }

    /**
     * 创建目录
     */
    public ResponseEntity mkdir(String fileSystem, int type, String path) {
        try {
            String[] split = path.split("/");
            mkdirParent(split, String.format("%s/%s", ZkConstant.META_DATA_PATH, fileSystem), "", 0);
            String metapath = String.format("%s/%s%s", ZkConstant.META_DATA_PATH, fileSystem, path);
            StatInfo statInfo = new StatInfo();
            statInfo.setMtime(DateUtil.date().getTime());
            statInfo.setSize(0);
            statInfo.setType(FileType.get(type));
            statInfo.setPath(path);
            if (curatorFramework.checkExists().forPath(metapath) == null) {
                curatorFramework.create().forPath(metapath, ByteConvertUtil.toByte(statInfo));
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return ResponseEntity.ok().build();
    }

    /**
     * 查看目录
     */
    public ResponseEntity listdir(String fileSystem, String path) {
        String metapath = String.format("%s/%s%s", ZkConstant.META_DATA_PATH, fileSystem, path);
        try {
            if (curatorFramework.checkExists().forPath(metapath) == null) {
                return ResponseEntity.status(500).body("路径不存在~");
            }
            List<String> list = curatorFramework.getChildren().forPath(metapath);
            List<StatInfo> statInfos = new ArrayList<>();
            for (String child : list) {
                StatInfo statInfo = getStatInfo(metapath + "/" + child);
                statInfos.add(statInfo);
            }
            return ResponseEntity.ok(statInfos);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return ResponseEntity.notFound().build();
    }

    /**
     * 删除文件
     */
    public ResponseEntity delete(String fileSystem, String path) {
        if ("".equals(path) || path == null) {
            return ResponseEntity.status(500).body("路径不存在~");
        }
        String metapath = String.format("%s/%s%s", ZkConstant.META_DATA_PATH, fileSystem, path);
        try {
            if (curatorFramework.checkExists().forPath(metapath) == null) {
                return ResponseEntity.status(500).body("路径不存在~");
            }
            if (curatorFramework.getChildren().forPath(metapath).size() != 0) {
                return ResponseEntity.status(500).body("该目录下不为空~");
            }
            StatInfo statInfo = getStatInfo(metapath);
            deleteData(statInfo, fileSystem, path);
            curatorFramework.delete().forPath(metapath);
            //查看数据所在的节点，全部删除
        } catch (Exception e) {
            e.printStackTrace();
            return ResponseEntity.status(500).build();
        }
        return ResponseEntity.ok().build();
    }

    /**
     * 保存文件写入成功后的元数据信息，包括文件path、size、三副本信息等
     */
    public ResponseEntity commitWrite(String fileSystem, MetaWrite metaWrite) {
        String metapath = String.format("%s/%s%s", ZkConstant.META_DATA_PATH, fileSystem, metaWrite.getPath());
        StatInfo statInfo = new StatInfo();
        statInfo.setReplicaData(metaWrite.getReplicaData());
        statInfo.setSize(metaWrite.getSize());
        statInfo.setMtime(DateUtil.date().getTime());
        statInfo.setType(FileType.File);
        statInfo.setPath(metaWrite.getPath());
        try {
            if (curatorFramework.checkExists().forPath(metapath) != null) {
                curatorFramework.setData().forPath(metapath, ByteConvertUtil.toByte(statInfo));
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return ResponseEntity.ok().build();
    }

    /**
     * 根据文件path查询三副本的位置，返回客户端具体ds、文件分块信息
     */
    public ResponseEntity open(String fileSystem, String path) {
        String metapath = String.format("%s/%s%s", ZkConstant.META_DATA_PATH, fileSystem, path);
        try {
            StatInfo statInfo = getStatInfo(metapath);
            List<ReplicaData> replicaData = statInfo.getReplicaData();
            return ResponseEntity.ok(replicaData);
        } catch (Exception e) {
            return ResponseEntity.notFound().build();
        }
    }

    private StatInfo getStatInfo(String metapath) throws Exception {
        byte[] bytes = curatorFramework.getData().forPath(metapath);
        return (StatInfo) ByteConvertUtil.toObj(bytes, StatInfo.class);
    }


    /**
     * 创建父级目录 - 保留最后一级不创建
     */
    private void mkdirParent(String[] child, String parent, String relPath, int idx) throws Exception {
        if (idx == child.length - 1) {
            return;
        }
        String path = parent;
        String nRelPath = relPath;
        if (!"".equals(child[idx])) {
            path = parent + "/" + child[idx];
            nRelPath = relPath + "/" + child[idx];
            if (curatorFramework.checkExists().forPath(path) == null) {
                StatInfo statInfo = new StatInfo();
                statInfo.setMtime(DateUtil.date().getTime());
                statInfo.setSize(0);
                statInfo.setType(FileType.Directory);
                statInfo.setPath(nRelPath);
                curatorFramework.create().forPath(path, ByteConvertUtil.toByte(statInfo));
            }
        }
        mkdirParent(child, path, nRelPath, idx + 1);
    }

    @Async
    public void deleteData(StatInfo statInfo, String fileSystem, String path) {
        List<ReplicaData> list = statInfo.getReplicaData();
        for (ReplicaData replicaData : list) {
            String durl = replicaData.dsNode;
            int size = statInfo.getSize();
             HashMap<String, String> map = new HashMap<>();
             map.put("path",path);
             map.put("size",size + "");
            FSHttpUtils.get(durl + DataServerApi.DELETE, map, fileSystem);
        }
    }


}
